package rocketmq

import (
	"context"
	"errors"
	"gitee.com/vrv_media/go-micro-framework/pkg/mq"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
	"sync"
)

// RocketMQTransactionProducer 基于rocket的事物消息的实现
type RocketMQTransactionProducer struct {
	addr           []string
	producerGroup  string
	listener       primitive.TransactionListener
	retry          int // 发送到 rocketmq 重试次数
	producerClient rocketmq.TransactionProducer
	once           sync.Once
}

func NewRocketMQTransactionProducer(addr []string, producerGroup string, listener primitive.TransactionListener) (mq.TransactionProducer, error) {
	return NewRocketMQTransactionProducerWithRetry(addr, producerGroup, listener, 3)
}

func NewRocketMQTransactionProducerWithRetry(addr []string, producerGroup string, listener primitive.TransactionListener, retry int) (mq.TransactionProducer, error) {
	if len(addr) == 0 {
		return nil, errors.New("addr is empty")
	}
	if listener == nil {
		return nil, errors.New("listener is empty")
	}
	return &RocketMQTransactionProducer{
		addr:          addr,
		producerGroup: producerGroup,
		listener:      listener,
		retry:         retry,
	}, nil
}

func (r *RocketMQTransactionProducer) ProducerStart(opts ...producer.Option) error {
	var err error

	r.once.Do(func() {
		var producerOpts []producer.Option
		if len(r.addr) > 0 {
			producerOpts = append(producerOpts, producer.WithNsResolver(
				primitive.NewPassthroughResolver(r.addr),
			))
		} else {
			err = errors.New("rocketmq Addr not found")
			return
		}

		if r.producerGroup != "" {
			producerOpts = append(producerOpts, producer.WithGroupName(r.producerGroup))
		}

		if r.retry > 0 {
			producerOpts = append(producerOpts, producer.WithRetry(r.retry))
		}
		if len(opts) > 0 {
			producerOpts = append(producerOpts, opts...)
		}

		r.producerClient, err = rocketmq.NewTransactionProducer(r.listener, producerOpts...)
		if err != nil {
			return
		}
		if err = r.producerClient.Start(); err != nil {
			return
		}
	})
	return err
}

func (r *RocketMQTransactionProducer) ProducerStop() error {
	if r.producerClient != nil {
		return r.producerClient.Shutdown()
	}
	return nil
}

func (r *RocketMQTransactionProducer) MessageInTransaction(topic string, data []byte, shardingKeyValue *string) error {
	msg := primitive.NewMessage(topic, data)
	if shardingKeyValue != nil && *shardingKeyValue != "" {
		msg.WithShardingKey(*shardingKeyValue)
	}
	_, err := r.producerClient.SendMessageInTransaction(context.Background(), msg)
	return err
}
